// @Author EthanScriptOn
// @Desc
package mysql_operator

import (
	"bufio"
	"fmt"
	"gitee.com/fatzeng/srf_switch_basic_components/common/identity"
	"gitee.com/fatzeng/srf_switch_expansion_package/extend_config"
	"gitee.com/fatzeng/srf_switch_stream_bridge/framework/actuator"
	"gitee.com/fatzeng/srf_switch_stream_bridge/framework/operator"
	"gitee.com/fatzeng/srf_switch_stream_bridge/framework/operator/op_example"
	"io"
	"os"
	"time"
)

type LocalReadOperator struct {
	*op_example.LoopDetectionMultiChannelDataOperator
	*identity.IdentityInformationCarrier
}

func (l *LocalReadOperator) ExecuteWithLoop(ctx *actuator.SrfSwitchBridgeActuatorContext, operatorConfigIdentifier interface{}, structureParam interface{}) (err error) {
	operatorIdentifier, err := l.IdentityInformationCarrier.GenerateUniqueIdentifier()
	if err != nil {
		_ = ctx.Record(actuator.OperatorExecute, fmt.Sprintf("Error generating unique identifier: %v", err))
		return err
	}
	pipeline := l.GetDataPipeline(operatorIdentifier)
	if pipeline == nil {
		_ = ctx.Record(actuator.OperatorExecute, fmt.Sprintf("Error getting data pipeline: %v", err))
		return err
	}
	config := operator.GetOperatorConfig(operatorConfigIdentifier)
	if config == nil {
		return fmt.Errorf("config is nil")
	}
	operatorConfig, ok := config.(*extend_config.LocalReadOperatorConfig)
	if !ok {
		return fmt.Errorf("config is not MysqlQueryOperatorConfig")
	}
	if !operatorConfig.IsValid() {
		return fmt.Errorf("MysqlQueryOperatorConfig is not valid")
	}
	contentChannel, errChannel := tailFile(operatorConfig.FilePath, operatorConfig.SleepDuration)

	for {
		select {
		case fileErr, ok := <-errChannel:
			if !ok {
				return fmt.Errorf("the err channel has been closed")
			} else {
				return fileErr
			}
		case content, ok := <-contentChannel:
			if !ok {
				return fmt.Errorf("the content channel has been closed")
			} else {
				pipeline <- content
			}
		}
	}
}

func tailFile(filename string, sleepDuration time.Duration) (contentChannel <-chan string, errChannel <-chan error) {
	fileContentChannel := make(chan string)
	fileErrorChannel := make(chan error)
	go func() {
		var offset int64 = 0
		for {
			file, openFileErr := os.Open(filename)
			if openFileErr != nil {
				fileErrorChannel <- fmt.Errorf("failed to open file: %v", openFileErr)
				return
			}
			_, err := file.Seek(offset, io.SeekStart)
			if err != nil {
				fileErrorChannel <- fmt.Errorf("failed to seek file: %v", err)
				return
			}
			reader := bufio.NewReader(file)
			var isPrefixCache bool
			var lineCache string
			var offsetCache int64
			for {
				line, isPrefix, err := reader.ReadLine()
				if err != nil && err != io.EOF {
					file.Close()
					fileErrorChannel <- fmt.Errorf("error reading line: %v", err)
					return
				}
				if !isPrefix {
					if len(line) == 0 {
						continue
					}
					if line[len(line)-1] != '~' {
						time.Sleep(2 * sleepDuration)
						file.Close()
						break
					}

					offset += int64(len(line)) + 1
					if isPrefixCache {
						offset += offsetCache
					}
					lineStr := string(line)
					if isPrefixCache {
						lineCache += lineStr
						lineStr = lineCache
					}
					isPrefixCache = false
					if lineStr != "" {
						fileContentChannel <- lineStr
					}
				} else {
					isPrefixCache = true
					lineCache += string(line)
					offsetCache += int64(len(line))
				}

				if err == io.EOF {
					file.Close()
					break
				}
			}
			time.Sleep(sleepDuration)
		}
	}()
	return fileContentChannel, fileErrorChannel
}
